package amq.utils;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

import amq.utils.enums.ELangCode;
import amq.utils.thds.JSFAgentAQThd;
import amq.utils.thds.JSFAgentSHBThd;
import flib.util.TimeStr;

public class JSFAgent {
	public int					TASK_WAIT_TIME = 2000;
	public String 				AQN = "WEB.ASW_Qx"; 	
	public String 				brokerURI = "";
	public int					taskID=1;
	
	
	ActiveMQConnectionFactory 	connectionFactory=null;
	Connection 					connection = null;
	Session 					session = null;
	JSFAgentAQThd 				aqt = null;
	JSFAgentSHBThd				ash = null;
	HashMap<String,Task> 		taskMap = new HashMap<String,Task>();
	HashMap<String,SrvStat>		servMap = new HashMap<String,SrvStat>();
	int							aliveSrvCnt=-1;

	public JSFAgent(String bui, String aqn)throws JMSException{
		this.AQN = aqn;
		this.brokerURI = bui;
		connectionFactory = new ActiveMQConnectionFactory(brokerURI);
		// Create a Connection
        connection = connectionFactory.createConnection();
        connection.start();

        // Create a Session
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        
        // Create answer queue thd
        aqt = new JSFAgentAQThd(AQN, brokerURI, this);
        new Thread(aqt).start();
        ash = new JSFAgentSHBThd(GlobalAMQ.SayHiTopic, brokerURI, this);
        new Thread(ash).start();
	}
	
	public boolean resetBroker(String bui)
	{		
		try
		{
			exit();
			Thread.sleep(5000);
			connectionFactory = new ActiveMQConnectionFactory(bui);
			// Create a Connection
	        connection = connectionFactory.createConnection();
	        connection.start();

	        // Create a Session
	        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	        
	        // Create answer queue thd
	        aqt = new JSFAgentAQThd(AQN, bui, this);
	        new Thread(aqt).start();
	        ash = new JSFAgentSHBThd(GlobalAMQ.SayHiTopic, bui, this);
	        new Thread(ash).start();
			this.brokerURI = bui;
			return true;
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		catch(Exception e)
		{
			e.printStackTrace();
		}
		return false;
	}
	
	public boolean _restartSession() throws JMSException
	{
		close();
		connection = connectionFactory.createConnection();
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		return true;
	}
	
	public List<Task> querySent(List<String> sents, String queryTopic, int timeout)
	{
		aliveSrvCnt = 0;
		for(SrvStat ss:servMap.values())
		{
			if(ss.isAlive && ss.topic.equals(queryTopic))
			{
				aliveSrvCnt++;
			}
		}
		System.out.printf("\t[Info] Alive Service Number(%s)=%d...\n", queryTopic, aliveSrvCnt);
		taskMap.clear();
		MessageProducer producer = null;
		Task task = null;
		List<Task> taskList = new ArrayList<Task>();
		if(aliveSrvCnt==0)
		{
			for(String sent:sents)
			{
				task = new Task("-", sent, System.currentTimeMillis());
				task.resp.add("No Service Available!");
				taskList.add(task);
			}
			return taskList;
		}
		try
		{
			// Create the destination (Topic or Queue)
            Destination destination = session.createTopic(queryTopic.trim());

            // Create a MessageProducer from the Session to the Topic or Queue
            producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            
            // Create a messages
            for(String sent:sents)
            {
            	String tidStr = String.format("T%d", taskID);
                task = new Task(tidStr, sent, System.currentTimeMillis());
                TextMessage message = null;
                message = session.createTextMessage(sent);
                message.setStringProperty("UID", "John");
            	message.setStringProperty("TID", tidStr.trim());
                message.setStringProperty("ReplyTo", AQN);
                message.setDoubleProperty("TimeStamp", new Date().getTime());
                
            	// Tell the producer to send the message
                producer.send(message);                                
                System.out.printf("\t[Info] Add task('%s') for sent='%s'...\n", tidStr, sent);                
                taskMap.put(tidStr.trim(), task);
                taskList.add(task);
                taskID=(taskID+1)%10000;
            }
            
            // Shortcut
            List<String> doneList = new ArrayList<String>();
            while(timeout>0)
            {
            	int to = 1000;
            	if(timeout>1000) timeout-=1000;
            	else 
            	{
            		to = timeout;
            		timeout = 0;
            	}
            	Thread.sleep(to);
            	Iterator<Task> iter = taskMap.values().iterator();
            	doneList.clear();
            	while(iter.hasNext())
            	{
            		Task t = iter.next();
            		if(aliveSrvCnt>0 && t.resp.size()==aliveSrvCnt) {
            			doneList.add(t.tid);            			
            		}
            	}
            	for(String tid:doneList)
            	{
            		taskMap.remove(tid);
            		System.out.printf("\t[Info] Remove Task=%s...(%d last)\n", tid, taskMap.size());
            	}
            	if(taskMap.size()==0) break;
            }
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		catch(Exception e)
		{
			e.printStackTrace();
		}
		finally
		{
			try{ if(producer!=null) producer.close();} catch(Exception e){}
		}
		return taskList;
	}
	
	public List<Task> querySent(List<String> sents, int timeout)
	{		
		return querySent(sents, GlobalAMQ.QueryTopic, timeout);
	}
	
	public Task querySent(String sent, int timeout)
	{
		return querySent(sent, GlobalAMQ.QueryTopic, timeout);
	}
	
	public Task querySent(String sent, String topic, int timeout)
	{
		taskMap.clear();
		MessageProducer producer = null;
		Task task = null;
		try
		{
			// Check alive services
			aliveSrvCnt = 0;
			for(SrvStat ss:servMap.values())
			{
				if(ss.isAlive && ss.topic.equals(topic))
				{
					aliveSrvCnt++;
				}
			}
			System.out.printf("\t[Info] Alive Service Number(%s)=%d...\n", topic, aliveSrvCnt);
			
			// Create the destination (Topic or Queue)
            Destination destination = session.createTopic(topic);

            // Create a MessageProducer from the Session to the Topic or Queue
            producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            
            // Create a messages
            String tidStr = String.format("T%d", taskID);
            task = new Task(tidStr, sent, System.currentTimeMillis());
            TextMessage message = null;
            message = session.createTextMessage(sent);
            message.setStringProperty("UID", "John");
        	message.setStringProperty("TID", tidStr);
            message.setStringProperty("ReplyTo", AQN);
            message.setDoubleProperty("TimeStamp", new Date().getTime());
            //System.out.println("Query Service: "+ lt + " : " + Thread.currentThread().getName());
        	// Tell the producer to send the message
            producer.send(message);                
            //TaskMap.put(tidStr, new Task(tidStr, line, System.currentTimeMillis()));
            taskID=(taskID+1)%10000;
            taskMap.put(tidStr, task);

            // Shortcut
            List<String> doneList = new ArrayList<String>();
            while(timeout>0)
            {
            	int to = 1000;
            	if(timeout>1000) timeout-=1000;
            	else 
            	{
            		to = timeout;
            		timeout = 0;
            	}
            	Thread.sleep(to);
            	Iterator<Task> iter = taskMap.values().iterator();
            	doneList.clear();
            	while(iter.hasNext())
            	{
            		Task t = iter.next();
            		if(aliveSrvCnt>0 && t.resp.size()==aliveSrvCnt) {
            			doneList.add(t.tid);            			
            		}
            	}
            	for(String tid:doneList)
            	{
            		taskMap.remove(tid);
            		System.out.printf("\t[Info] Remove Task=%s...(%d last)\n", tid, taskMap.size());
            	}
            	if(taskMap.size()==0) break;
            }
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		catch(Exception e)
		{
			e.printStackTrace();
		}
		finally
		{
			try{ if(producer!=null) producer.close();} catch(Exception e){}
		}
		return task;
	}
	
	/**
	 * BD: Heart Beat to service
	 * @param lt: Language Type(Trad,Simp,Engi)
	 * @param timeout: Timeout to wait response
	 */
	public void serviceHB(ELangCode lc, int timeout)
	{
		MessageProducer producer = null;
		try
		{
			for(SrvStat ss:servMap.values()) ss.isAlive = false;
			// Create the destination (Topic or Queue)
            Destination destination = session.createTopic(GlobalAMQ.ServiTopic);

            // Create a MessageProducer from the Session to the Topic or Queue
            producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            
            // Create a messages            
            TextMessage message = null;
            message = session.createTextMessage("");
            message.setStringProperty("TID", String.format("S%d", System.currentTimeMillis()));
            message.setStringProperty("ReplyTo", AQN);
            if(lc!=null) message.setStringProperty("LangCode", lc.toString());
            //System.out.println("Query Service: "+ lt + " : " + Thread.currentThread().getName());
        	// Tell the producer to send the message
            producer.send(message);                
            Thread.sleep(timeout);
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		catch(Exception e)
		{
			e.printStackTrace();
		}
		finally
		{
			try{ if(producer!=null) producer.close();} catch(Exception e){}
		}
		aliveSrvCnt = 0;
		for(SrvStat ss:servMap.values())
		{
			if(!ss.isAlive)
			{
				System.out.printf("\t[Warn] Service='%s' is out of service...\n", ss.name);
				ss.latestDieTS = System.currentTimeMillis();
			}
			else aliveSrvCnt++;
		}
	}
	
	public void close() throws JMSException
	{
		if(session!=null) session.close();
        if(connection!=null) connection.close();        
	}
	
	public void exit() throws JMSException
	{
		close();
		if(aqt!=null)
		{
			aqt.isStop = true;
			aqt = null;
		}
		if(ash!=null)
		{
			ash.isStop = true;
			ash = null;
		}
	}
	
	public void logQuery(String sent_id, String topic, String user_name, String sent, String resp)
	{
		logQuery(sent_id, topic, user_name, sent, resp, "localhost");
	}
	
	public void logQuery(String sent_id, String topic, String user_name, String sent, String resp, String from)
	{
		MessageProducer producer = null;
		try
		{						
			// Create the destination (Topic or Queue)
            Destination destination = session.createTopic(GlobalAMQ.LogQueue);

            // Create a MessageProducer from the Session to the Topic or Queue
            producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            
            // Create a messages           
            TextMessage message = null;
            message = session.createTextMessage("query");
            message.setStringProperty("sent_id", sent_id);
        	message.setStringProperty("user_name", user_name);
            message.setStringProperty("sent", sent);
            message.setStringProperty("resp", resp);
            message.setStringProperty("topic", topic);
            message.setStringProperty("from", from);
            //System.out.println("Query Service: "+ lt + " : " + Thread.currentThread().getName());
        	// Tell the producer to send the message
            System.out.printf("\t[Info] Save Query Log: %s...\n", message);
            producer.send(message);                            
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		catch(Exception e)
		{
			e.printStackTrace();
		}
		finally
		{
			try{ if(producer!=null) producer.close();} catch(Exception e){}
		}
	}
	
	public void logFeedback(String user_name, String sent_id, String suggestion)
	{
		logFeedback(user_name, sent_id, suggestion, "localhost");
	}
	
	public void logFeedback(String user_name, String sent_id, String suggestion, String from)
	{
		MessageProducer producer = null;
		try
		{						
			// Create the destination (Topic or Queue)
            Destination destination = session.createTopic(GlobalAMQ.LogQueue);

            // Create a MessageProducer from the Session to the Topic or Queue
            producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            
            // Create a messages           
            TextMessage message = null;
            message = session.createTextMessage("feedback");
            message.setStringProperty("sent_id", sent_id);
        	message.setStringProperty("suggestion", suggestion); 
        	message.setStringProperty("name", user_name);
        	message.setStringProperty("from", from);
            System.out.printf("\t[Info] Save Feedback: %s...\n", message);
            producer.send(message);                            
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		catch(Exception e)
		{
			e.printStackTrace();
		}
		finally
		{
			try{ if(producer!=null) producer.close();} catch(Exception e){}
		}
	}
	
	public synchronized void srvmCallBack(String tid, String resp, String topic)
	{
		System.out.printf("\t[Info] Service='%s' Heart beat...\n", resp);
		SrvStat ss = servMap.get(resp);
		if(ss==null) {
			ss = new SrvStat(resp);			
			servMap.put(resp, ss);			
		}
		ss.isAlive=true;
		ss.latestAliveTS = System.currentTimeMillis();
		ss.latestDieTS = -1;
		ss.topic = topic;
	}
	
	public synchronized void taskCallBack(String tid, String resp)
	{
		Task task = taskMap.get(tid.trim());
		if(task!=null) {
			System.out.printf("\t[Info] Update task(%s)=%s\n", tid, resp);
			task.resp.add(resp);
			task.lastRTS = System.currentTimeMillis();
			task.isReady=true;
		} else
		{
			System.err.printf("\t[Error] Unknown TaskID='%s'!!!\n", tid);
		}
	}
	
	public synchronized void srvhCallback(String sn, String say, String topic)
	{
		System.out.printf("\t[Info] Service='%s'(%s) say %s...\n", sn, topic, say);
		SrvStat sStat = servMap.get(sn);
		if(sStat==null) {
			sStat = new SrvStat(sn);
			servMap.put(sn, sStat);			
		}
		if(say.trim().equals("hi"))
		{
			sStat.isAlive=true;
			sStat.latestAliveTS = System.currentTimeMillis();
			sStat.latestDieTS = -1;
			sStat.topic = topic;
			/*if(aliveSrvCnt>=0)
			{
				aliveSrvCnt+=1;
			}
			else
			{
				aliveSrvCnt = 0;
				for(SrvStat ss:servMap.values())
				{
					if(!ss.isAlive)
					{
						System.out.printf("\t[Warn] Service='%s' is out of service...\n", ss.name);
						ss.latestDieTS = System.currentTimeMillis();
					}
					else aliveSrvCnt++;
				}
			}*/
		}
		else if(say.equals("bye"))
		{
			sStat.isAlive=false;
			sStat.topic = topic;
			sStat.latestDieTS = System.currentTimeMillis();
			/*if(aliveSrvCnt>0)
			{
				aliveSrvCnt-=1;
			}
			else
			{
				aliveSrvCnt = 0;
				for(SrvStat ss:servMap.values())
				{
					if(!ss.isAlive)
					{
						System.out.printf("\t[Warn] Service='%s' is out of service...\n", ss.name);
						ss.latestDieTS = System.currentTimeMillis();
					}
					else aliveSrvCnt++;
				}
			}*/
		}			
		//System.out.printf("\t[Info] %d alive services...\n", aliveSrvCnt);
	}
	
	public static class SrvStat{
		public String 		name;
		public long			latestAliveTS;
		public boolean		isAlive;
		public long			latestDieTS;
		public String 		topic;
		public SrvStat(String name)
		{
			this.name = name;
			this.latestAliveTS = System.currentTimeMillis();
			isAlive = true;
		}
	}
	
	public static class Task{
		public long			timeStamp;
		public long			lastRTS;
		public String 		tid;
		public String 		sent;
		public boolean		isReady = false;
		public List<String> resp = new ArrayList<String>();
		
		public Task(String tid, String sent, long ts){this.tid = tid; this.sent = sent; this.timeStamp = ts;}
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception{
		String sents[] = {"今天要到国际会议中心",
				          "洽谈一笔户外广告的生意",
				          "他操着砂哑的嗓音唱了一首歌", 
				          "新同事跟大家相处得非常融恰",
				          "这个沙窗很漂亮",
				          "这里还要拿砂轮打磨一下"};
		String tradSent = "結婚以後他們很恩情";
		List<String> sentList = new ArrayList<String>();
		for(String sent:sents) sentList.add(sent);
		JSFAgent jsfAgent = new JSFAgent("tcp://localhost:61616", "WEB.ASW_Test"); // "tcp://140.112.31.76:61616"
		//System.out.printf("\t[Info] Switch broker...\n");
		//jsfAgent.resetBroker("tcp://localhost:61616");
		System.out.printf("\t[Info] Query Service...\n");
		jsfAgent.serviceHB(ELangCode.TraditionChinese, 3000);
		//Thread.sleep(2000);
		System.out.printf("\t[Info] Query Sent='%s'...\n", tradSent);
		Task task = jsfAgent.querySent(tradSent, GlobalAMQ.QueryTradTopic, 5000);
		int sc = 0;
		String sid = String.format("S%d_%d", System.currentTimeMillis(), sc);
		if(task.resp.size()==0) System.out.printf("\t[Info] Sent='%s' has no response!\n", tradSent);
		else
		{
			System.out.printf("\t[Info] Sent='%s':\n", tradSent);
			StringBuffer respBuf = new StringBuffer();
			for(String r:task.resp) {
				System.out.printf("\t\t%s\n", r);
				respBuf.append(String.format("%s\r\n", r));
			}
			jsfAgent.logQuery(sid, GlobalAMQ.QueryTradTopic, "john", tradSent, respBuf.toString());
			jsfAgent.logFeedback("guest", sid, "Good");
		}
		//Thread.sleep(2000);
		jsfAgent.serviceHB(ELangCode.SimpleChinese, 3000);
		long st = System.currentTimeMillis();
		System.out.printf("\t[Info] Query Multiple Sentences(%d)...\n", sents.length);
		List<Task> taskList = jsfAgent.querySent(sentList, 5000);
		for(int i=0; i<sentList.size(); i++)
		{
			Task t = taskList.get(i);
			String sent = sentList.get(i);
			if(t.resp.size()>0)
			{
				System.out.printf("\tSent='%s':\n", sent);
				for(String r:t.resp) System.out.printf("\t\t%s\n", r);
			}
			else{
				System.out.printf("\tSent='%s' has no response!\n", sent);
			}
		}
		System.out.printf("\t[Info] Multiple sentence query takes %s!\n", TimeStr.ToStringFrom(st));
		
		System.in.read();
		jsfAgent.exit();		
		System.out.printf("\t[Info] Exit main...\n");
	}
}
